package com.mjf.gmall.realtime.dim

import com.alibaba.fastjson.{JSON, JSONObject}
import com.mjf.gmall.realtime.bean.dim.SkuInfo
import com.mjf.gmall.realtime.util.{MyKafkaUtil, OffsetManager, PhoenixUtil}
import org.apache.hadoop.conf.Configuration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.phoenix.spark._
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 实时维护商品sku维度表，关联category3、trademark、spuInfo表生成大宽表
 *
 * 技术栈：
 * zookeeper/kafka/redis/hadoop/HBase/phoenix
 *
 * phoenix建表语句：
 * create table gmall0105_sku_info(id varchar primary key, spu_id  varchar, price  varchar, sku_name varchar, tm_id varchar, category3_id varchar, create_time varchar, category3_name varchar, spu_name varchar, tm_name varchar) SALT_BUCKETS = 3;
 *
 * SPU：苹果6（商品聚合信息的最小单位），如手机->苹果手机->苹果6，苹果6就是SPU。
 * SKU：土豪金 16G 苹果6 （商品的不可再分的最小单元）。
 */
object SkuInfoApp {
  def main(args: Array[String]): Unit = {

    // 读取Kafka中商品sku的topic
    // 加载流
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("SkuInfoApp")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))

    val topic = "ods_sku_info"
    val groupId = "sku_info_group"

    // 从redis读取offset
    val kafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)

    // 从Kafka消费数据
    var recordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (kafkaOffsetMap == null && kafkaOffsetMap.size > 0) {
      // 如果redis中已经有了offset信息
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, kafkaOffsetMap, groupId)
    } else {
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }

    //获得偏移结束点
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange] // 写在算子外面为启动时执行
    val inputGetOffsetDstream: DStream[ConsumerRecord[String, String]] = recordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 写在算子里面是周期性执行（driver端）
        rdd
    }


    // 转换格式，将 JSON 转为 SkuInfo
    val skuInfoDStream: DStream[SkuInfo] = inputGetOffsetDstream.map {
      record =>
        val jsonString: String = record.value()
        val skuInfo: SkuInfo = JSON.parseObject(jsonString, classOf[SkuInfo])
        skuInfo
    }

    // 关联category3、trademark、spuInfo表生成大宽表
    val skuInfoWideDStream: DStream[SkuInfo] = skuInfoDStream.mapPartitions {
      skuInfoIter =>
        val skuInfoList: List[SkuInfo] = skuInfoIter.toList

        if (skuInfoList.size > 0) {
          val category3Id: String = skuInfoList.map(_.category3_id).mkString("','")
          val tmId: String = skuInfoList.map(_.tm_id).mkString("','")
          val spuId: String = skuInfoList.map(_.spu_id).mkString("','")
          // 查询categroy3信息
          val categroy3SQL: String = "select ID,NAME from GMALL0105_BASE_CATEGORY3 where ID in ('" + category3Id + "')"
          val categroy3List: List[JSONObject] = PhoenixUtil.queryList(categroy3SQL)
          val categroy3Map: Map[String, JSONObject] = categroy3List.map(category3 => (category3.getString("ID"), category3)).toMap
          // 查询trademark信息
          val trademarkSQL: String = "select ID,TM_NAME from GMALL0105_BASE_TRADEMARK where ID in ('" + tmId + "')"
          val trademarkList: List[JSONObject] = PhoenixUtil.queryList(trademarkSQL)
          val trademarkMap: Map[String, JSONObject] = trademarkList.map(trademark => (trademark.getString("ID"), trademark)).toMap
          // 查询spuInfo信息
          val spuInfoSQL: String = "select ID,SPU_NAME from GMALL0105_SPU_INFO where ID in ('" + spuId + "')"
          val spuInfoList: List[JSONObject] = PhoenixUtil.queryList(spuInfoSQL)
          val spuInfoMap: Map[String, JSONObject] = spuInfoList.map(spuInfo => (spuInfo.getString("ID"), spuInfo)).toMap

          for (skuInfo <- skuInfoList) {
            val category3Obj: JSONObject = categroy3Map.getOrElse(skuInfo.category3_id, null)
            val trademarkObj: JSONObject = trademarkMap.getOrElse(skuInfo.tm_id, null)
            val spuInfoObj: JSONObject = spuInfoMap.getOrElse(skuInfo.spu_id, null)
            if (category3Obj != null) {
              skuInfo.category3_name = category3Obj.getString("NAME") // 补充三级分类信息信息
            }
            if (trademarkObj != null) {
              skuInfo.tm_name = trademarkObj.getString("TM_NAME") // 补充品牌信息
            }
            if (spuInfoObj != null) {
              skuInfo.spu_name = spuInfoObj.getString("SPU_NAME") // 补充商品spu信息
            }
          }
        }

        skuInfoList.toIterator
    }

    // 将结果写入HBase
    skuInfoWideDStream.foreachRDD {
      skuInfoRDD =>
        skuInfoRDD.saveToPhoenix(
          "GMALL0105_SKU_INFO",
          Seq("ID", "SPU_ID", "PRICE", "SKU_NAME", "TM_ID", "CATEGORY3_ID", "CREATE_TIME", "CATEGORY3_NAME", "SPU_NAME", "TM_NAME"),
          new Configuration,
          Some("hadoop102,hadoop103,hadoop104:2181")
        )

        // 数据处理结束，偏移量提交到redis
        OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }

    skuInfoWideDStream.print(1000)

    ssc.start()
    ssc.awaitTermination()

  }
}
